﻿using Castle.Core.Logging;
using System;
using System.Collections.Generic;
using System.Text;
using Thrift.Protocol;
using Thrift.Transport;

namespace Inovout.YunYao.Cattle.WindowsService
{
    /// <summary>
    /// 持久化到HBase
    /// </summary>
    public class DataConveyor
    {
        public string HBaseIP { get; private set; }
        public int HBasePort { get; private set; }
        public string HBaseTableNamePrefix { get; private set; }
        private ILogger logger;
        private Dictionary<string, byte[]> monitorColumnDictionary;
        private TTransport transport;
        private Hbase.Client client;
        public DataConveyor(string hbaseIP, int hbasePort, string hbaseTableNamePrefix)
        {
            logger = ComponentRegistry.Resolve<ILogger>();
            HBaseIP = hbaseIP;
            HBasePort = hbasePort;
            HBaseTableNamePrefix = hbaseTableNamePrefix;
            monitorColumnDictionary = new Dictionary<string, byte[]> 
            { 
                 {"value",Encoding.UTF8.GetBytes("monitor:value")},
                 {"timestamp",Encoding.UTF8.GetBytes("monitor:timestamp")},
                 {"location",Encoding.UTF8.GetBytes("monitor:location")}
            };
            transport = new TSocket(HBaseIP, HBasePort);
            TProtocol tProtocol = new TBinaryProtocol(transport);
            client = new Hbase.Client(tProtocol);

        }
        /// <summary>
        /// 需增加时间戳参数，可参照下面URL处理。
        /// http://blog.csdn.net/chinalgf/article/details/8541133
        /// </summary>
        public void Push(string[] items)
        {
            try
            {
                if (!transport.IsOpen)
                {
                    transport.Open();
                }
                Dictionary<string, List<BatchMutation>> rowBatchDictionary = new Dictionary<string, List<BatchMutation>>();
                foreach (var item in items)
                {
                    Dictionary<string, string> kvDictionary = new Dictionary<string, string>();
                    foreach (var kv in item.Split('&'))
                    {
                        string[] kvArray = kv.Split('=');
                        kvDictionary.Add(kvArray[0], kvArray[1]);
                    }
                    BatchMutation rowBatchMutation = new BatchMutation();
                    rowBatchMutation.Row = Encoding.UTF8.GetBytes(kvDictionary["node"]);//测点代码
                    string tableName = HBaseTableNamePrefix + kvDictionary["node"].Substring(0, 3);
                    kvDictionary.Remove("node");
                    rowBatchMutation.Mutations = new List<Mutation>();
                    foreach (var kv in kvDictionary)
                    {
                        byte[] valueBuffer = Encoding.UTF8.GetBytes(kv.Value);
                        byte[] columnBuffer;
                        if (!monitorColumnDictionary.TryGetValue(kv.Key, out columnBuffer))
                        {
                            columnBuffer = Encoding.UTF8.GetBytes("solution:" + kv.Key);
                        }
                        rowBatchMutation.Mutations.Add(new Mutation
                        {
                            Column = columnBuffer,
                            Value = valueBuffer
                        });
                    }
                    List<BatchMutation> batchMutations;
                    if (!rowBatchDictionary.TryGetValue(tableName, out batchMutations))
                    {
                        batchMutations = new List<BatchMutation>();
                        rowBatchDictionary.Add(tableName, batchMutations);
                    }
                    batchMutations.Add(rowBatchMutation);
                }
                foreach (var kv in rowBatchDictionary)
                {
                    client.mutateRows(UTF8Encoding.UTF8.GetBytes(kv.Key), kv.Value, null);
                }
            }
            catch (Exception ex)
            {
                logger.Error("插入HBase报错", ex);
                AlertManager.Send(this.GetType().Name, ex);
            }
            finally
            {
                transport.Close();
            }
        }

    }
}
